package com.rainman.simpleredis.core.process_flow;


import cn.hutool.json.JSONArray;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.rainman.simpleredis.core.config.SimpleRedisConfig;
import com.rainman.simpleredis.core.distributed_lock.SimpleDistributedLock;
import com.rainman.simpleredis.core.scheduled.ScheduledInfo;
import com.rainman.simpleredis.core.scheduled.SimpleScheduled;
import com.rainman.simpleredis.core.scheduled.SimpleScheduledWorkerInterface;
import com.rainman.simpleredis.core.utils.DelayList;

import java.time.Duration;
import java.util.HashMap;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;


/**
 *
 * 分布式工作流,TC模式
 * 基于redis的消息发布订阅机制
 * T为尝试，C为取消回滚
 * 如A系统下单->B系统扣库存。如果B系统扣库存失败，先回滚B系统取消扣库存，后回滚A系统关闭订单
 *
 */
public class SimpleProcessFlow {

    private final SimpleRedisConfig simpleRedisConfig;
    private final SimpleDistributedLock simpleDistributedLock;

    private final DelayList delayList;
    private final ScheduledInfo scheduledInfo=ScheduledInfo.buildRate(23100L);
    private final HashMap<String,SimpleProcessFlowListener> listenerHashMap;

    private  final Long RecordDeadTime=1000L*3600*24;

    public final String ProcessFlowTigOriHead;

    public final String ProcessFlowHead;//源数据头 SR0.Ori.ProcessFlow:
    public final String ProcessFlowRecordHead;//保留数据头 SR0.Record.ProcessFlow:
    public final String ProcessFlowDeadHead;//死信数据头 SR0.Dead.ProcessFlow:
    public final String ProcessFlowTigHead;//触发往下进行 SR0.Tig.ProcessFlow:
    public final String ProcessFlowDoingTigHead;//触发通知进行中 SR0.Tig.ProcessFlow.Doing:
    public final String ProcessFlowSuccTigHead;//触发通知完成成功 SR0.Tig.ProcessFlow.Succ:
    public final String ProcessFlowErrTigHead;//触发通知失败错误 SR0.Tig.ProcessFlow.Err:

    public final String TigListenerTimeOutHead;//处理监听器过期 SR0.Tig.ProcessFlow.ListenerTimeOut:

    //分布式锁头数据
   // public  final String ScanDataLockHead;//SR0.Dlock.ProcessFlowScanData
    public  final String DealLockHead="SimpleRedis.ProcessFlowDealOne:";//单个任务处理锁头 SR0.Dlock.ProcessFlowDealOne:

    public SimpleProcessFlow(SimpleRedisConfig simpleRedisConfig
            , SimpleDistributedLock simpleDistributedLock
            , SimpleScheduled simpleScheduled
    ){
        this.simpleRedisConfig = simpleRedisConfig;
        this.simpleDistributedLock=simpleDistributedLock;
        delayList=new DelayList();
        listenerHashMap=new HashMap<>();

        String keyHead=simpleRedisConfig.getKeyHead()+ simpleRedisConfig.getProduct()+".";//SR0.
        ProcessFlowTigOriHead=keyHead+"Tig.ProcessFlow";

        ProcessFlowHead=keyHead+"Ori.ProcessFlow:";//SR0.Ori.ProcessFlow:
        ProcessFlowRecordHead=keyHead+"Record.ProcessFlow:";//SR0.Record.ProcessFlow:
        ProcessFlowTigHead=keyHead+"Tig.ProcessFlow:";//SR0.Tig.ProcessFlow:

        ProcessFlowDoingTigHead=keyHead+"Tig.ProcessFlow.Doing:";//SR0.Tig.ProcessFlow.Doing:
        ProcessFlowSuccTigHead=keyHead+"Tig.ProcessFlow.Succ:";//SR0.Tig.ProcessFlow.Succ:
        ProcessFlowErrTigHead=keyHead+"Tig.ProcessFlow.Err:";//SR0.Tig.ProcessFlow.Err:

        TigListenerTimeOutHead=keyHead+"Tig.ProcessFlow.ListenerTimeOut:";//SR0.Tig.ProcessFlow.ListenerTimeOut:

        ProcessFlowDeadHead=keyHead+"Dead.ProcessFlow:";//SR0.Dead.ProcessFlow:


        simpleScheduled.addWorker(new SimpleScheduledWorkerInterface() {
                    @Override
                    public void doWork() {
                        scanData();
                    }
                    @Override
                    public String key() {
                        return "SimpleProcessFlow.ScanData";
                    }
                    @Override
                    public ScheduledInfo info() {
                        return scheduledInfo;
                    }
                });

    }


    /**
     * 兜底数据扫描
     *
     */
    private void scanData() {

        Set<String> keys = simpleRedisConfig.getRedisInterface().keys(ProcessFlowHead + "*");//SR0.Ori.ProcessFlow:*

        for(String key:keys){//SR0.Ori.ProcessFlow:任务唯一键

            String value = simpleRedisConfig.getRedisInterface().getValue(key);
            ProcessFlowMsgBean processFlow = JSONUtil.toBean(value, ProcessFlowMsgBean.class);

            if(ProcessFlowMsgBean.StateDoIng.equals(processFlow.getState())){

                //进行中的，留意是否过期了
                if(processFlow.getOutTime()<System.currentTimeMillis()){
                    processFlow.setState(ProcessFlowMsgBean.StateUnDoIng);
                    simpleRedisConfig.getRedisInterface().set(key,JSONUtil.toJsonStr(processFlow));
                }

            }else if(ProcessFlowMsgBean.StateUnDoIng.equals(processFlow.getState())){

                //回滚中的，留意是否中断
                Boolean aBoolean = simpleDistributedLock.hasLock(DealLockHead + processFlow.getKey());

                if(!aBoolean){
                    tigNextStep(processFlow.getKey());
                }

            }

        }

    }


    public void addOne(ProcessFlowBuilder builder,SimpleProcessFlowListener listener){

        if(builder.getKey().contains(":")){
            throw new RuntimeException("key不能包含:");
        }

        if(listener!=null){
            listenerHashMap.put(builder.getKey(),listener);
            simpleRedisConfig.getRedisInterface().set(TigListenerTimeOutHead+builder.getKey(),""
                    , Duration.ofMillis(builder.getDur()));
        }

        ProcessFlowMsgBean processFlow=new ProcessFlowMsgBean();
        processFlow.setKey(builder.getKey());
        processFlow.setState(ProcessFlowMsgBean.StateDoIng);
        processFlow.setCurIndex(0);
        processFlow.setOutTime(System.currentTimeMillis()+builder.getDur());
        processFlow.setParamInfo(builder.getParamInfo());

        Boolean aBoolean = simpleRedisConfig.getRedisInterface().hasKey(ProcessFlowHead + builder.getKey());

        if(aBoolean){
            throw new RuntimeException("有重复key任务在进行");
        }

        for(String typ:builder.getWork()){
            processFlow.add(typ);
        }

        simpleRedisConfig.getRedisInterface().set(ProcessFlowHead + builder.getKey(),JSONUtil.toJsonStr(processFlow));

        tigNextStep(builder.getKey());//触发下一步

        if(listener!=null){
            simpleRedisConfig.getRedisInterface().set(TigListenerTimeOutHead+builder.getKey(),""
                    , Duration.ofMillis(builder.getDur()));
        }

    }


    /**
     * 发布订阅中获取消息
     * 并处理，下一步触发
     *
     * @param key
     */
    public void dealOne(String key){

        if(key.startsWith(ProcessFlowTigHead)){//下一步处理触发 SR0.Tig.ProcessFlow:任务唯一键
            doDeal(key);

        }else if(key.startsWith(ProcessFlowDoingTigHead)){//过程进行中完成时触发 SR0.Tig.ProcessFlow.Doing:任务唯一键:当前处理的下标:当前处理的类型名

            String[] split = key.split(":");
            SimpleProcessFlowListener simpleProcessFlowListener = listenerHashMap.get(split[1]);
            if(simpleProcessFlowListener!=null){
                simpleProcessFlowListener.onDoing(split[1],Integer.parseInt(split[2]),split[3]);
            }

        }else if(key.startsWith(ProcessFlowSuccTigHead)){//成功触发 SR0.Tig.ProcessFlow.Succ:任务唯一键

            String[] split = key.split(":");

            SimpleProcessFlowListener simpleProcessFlowListener = listenerHashMap.get(split[1]);

            if(simpleProcessFlowListener!=null){
                String value = simpleRedisConfig.getRedisInterface().getValue(ProcessFlowRecordHead+split[1]);
                ProcessFlowMsgBean processFlow = JSONUtil.toBean(value, ProcessFlowMsgBean.class);
                simpleProcessFlowListener.onSucc(split[1],processFlow.getReturnInfo());
                listenerHashMap.remove(split[1]);
            }

        }else if(key.startsWith(ProcessFlowErrTigHead)){//失败触发 SR0.Tig.ProcessFlow.Err:任务唯一键

            String[] split = key.split(":");

            SimpleProcessFlowListener simpleProcessFlowListener = listenerHashMap.get(split[1]);
            if(simpleProcessFlowListener!=null){
                simpleProcessFlowListener.onErr(split[1],split[2]);
                listenerHashMap.remove(split[1]);
            }

        }

    }


    /**
     * 下一步 处理执行
     * @param key
     */
    public void doDeal(String key){

        //真实的key
        key=key.split(ProcessFlowTigHead)[1];

        String value = simpleRedisConfig.getRedisInterface().getValue(ProcessFlowHead + key);

        ProcessFlowMsgBean processFlow = JSONUtil.toBean(value, ProcessFlowMsgBean.class);

        String workerTypeName = processFlow.getTypes().get(processFlow.getCurIndex());

        SimpleProcessFlowWorkerInterface workerInterface =
                simpleRedisConfig.getFlowWorkerInterfaceHashMap().get(workerTypeName);

        if(workerInterface==null){//本系统没有消费者
            delayList.add(key);
            return;
        }

        //抢之前看下自己状态,上次有异常状态的让20ms。尽可能让其他消费者抢到
        boolean b = delayList.hasKey(key);
        if(b){
            try {
                Thread.sleep(20L);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        //抢处理权限
        Boolean dLockAdd = simpleDistributedLock.dLockAdd(DealLockHead + key, 40000L);
        if(!dLockAdd){
            return;
        }

        if(b){
            delayList.remove(key);
        }

        //执行业务
        if(ProcessFlowMsgBean.StateDoIng.equals(processFlow.getState())){

            doingWork(processFlow,workerInterface);//正常流程处理

        }else if(ProcessFlowMsgBean.StateUnDoIng.equals(processFlow.getState())){

            unDoingWork(processFlow,workerInterface);//回滚流程处理

        }

    }

    /**
     * 正常流程处理
     *
     * @param processFlow
     * @param workerInterface
     */
    private void doingWork(ProcessFlowMsgBean processFlow, SimpleProcessFlowWorkerInterface workerInterface)  {

        ProcessWorkerReturn processWorkerReturn =new ProcessWorkerReturn(ProcessWorkerReturn.StateRetry);

        try{

            if(processFlow.getOutTime()<System.currentTimeMillis()){//已过期

                processWorkerReturn.setState(ProcessWorkerReturn.StateErr);
                processWorkerReturn.setReturnOrFailMsg("TimeOut");
                return;
            }

            String params = processFlow.getParamInfo().get(processFlow.getCurIndex());
            List<String> strings = JSONUtil.toList(params, String.class);

            processWorkerReturn = CompletableFuture.supplyAsync(()-> workerInterface.doWork(strings))
                    .get(processFlow.getOutTime()-System.currentTimeMillis()
                    , TimeUnit.MILLISECONDS);

            if(processWorkerReturn ==null){
                processWorkerReturn =new ProcessWorkerReturn(ProcessWorkerReturn.StateSuc);
            }

            if(ProcessWorkerReturn.StateSuc.equals(processWorkerReturn.getState())){//成功的话要处理参数

                processFlow.getReturnInfo().add(
                        processWorkerReturn.getReturnOrFailMsg()==null?"": processWorkerReturn.getReturnOrFailMsg());

                if(processFlow.getCurIndex()<processFlow.getTypes().size()-1){//未走到终点

                    String nextParams = processFlow.getParamInfo().get(processFlow.getCurIndex()+1);//处理下一步参数

                    JSONArray jsonArray = JSONUtil.parseArray(nextParams);


                    for(int i=0;i<jsonArray.size();i++){
                        String str = jsonArray.getStr(i);
                        if(str.startsWith("#{")){
                            //0.id
                            String s1 = str.replaceFirst("#\\{", "").replaceFirst("}", "");

                            //split[0]=0
                            String[] split = s1.split("\\.");

                            String targetReturnInfo = processFlow.getReturnInfo().get(Integer.parseInt(split[0]));

                            if(split.length==1){//参数类似为#{0}
                                jsonArray.set(i,targetReturnInfo);
                            }else {//参数类似为#{0.data.id}

                                JSONObject jsonObject = JSONUtil.parseObj(targetReturnInfo);

                                for(int j=1;j<split.length;j++){
                                    if(j== split.length-1){
                                        jsonArray.set(i,jsonObject.getStr(split[i]));
                                    }else {
                                        jsonObject=jsonObject.getJSONObject(split[i]);
                                    }
                                }

                            }
                        }
                    }


                    processFlow.getParamInfo().set(processFlow.getCurIndex()+1,jsonArray.toString());
                    processFlow.setCurIndex(processFlow.getCurIndex()+1);

                }else {//走到终点

                    processFlow.setState(ProcessFlowMsgBean.StateSucc);
                }

            }
            //执行错误，或者参数处理错误。都回滚 或 死信

        }catch (Exception e){
            e.printStackTrace();

            if(processWorkerReturn ==null){
                processWorkerReturn =new ProcessWorkerReturn(ProcessWorkerReturn.StateErr);
            }
            processWorkerReturn.setState(ProcessWorkerReturn.StateErr);
            processWorkerReturn.setReturnOrFailMsg(e.toString());

            throw new RuntimeException(e);

        }finally {


            if(ProcessWorkerReturn.StateSuc.equals(processWorkerReturn.getState())){

                if(ProcessFlowMsgBean.StateSucc.equals(processFlow.getState())){//已走到终点,成功执行完毕

                    //原始信息设置过期时间
                    simpleRedisConfig.getRedisInterface().delete(ProcessFlowHead+processFlow.getKey());
                    simpleRedisConfig.getRedisInterface().set(ProcessFlowRecordHead+processFlow.getKey()
                            ,JSONUtil.toJsonStr(processFlow),Duration.ofMillis(RecordDeadTime));


                    tigSucc(processFlow.getKey());//触发完全成功

                    simpleDistributedLock.dLockRemove(DealLockHead+processFlow.getKey());

                }else {//未走到终点，中途步骤成功

                    //更新数据
                    simpleRedisConfig.getRedisInterface().set(ProcessFlowHead + processFlow.getKey()
                            ,JSONUtil.toJsonStr(processFlow));

                    tigDoing(processFlow.getKey(),processFlow.getCurIndex()-1
                            ,processFlow.getTypes().get(processFlow.getCurIndex()-1));//Doing回调触发

                    simpleDistributedLock.dLockRemove(DealLockHead+processFlow.getKey());
                    tigNextStep(processFlow.getKey());//下一步触发

                }

            }else if(ProcessWorkerReturn.StateRetry.equals(processWorkerReturn.getState())){

                delayList.add(processFlow.getKey());

                simpleDistributedLock.dLockRemove(DealLockHead+processFlow.getKey());
                tigNextStep(processFlow.getKey());//下一步触发

            }else {//有节点处理失败的

                processFlow.setState(ProcessFlowMsgBean.StateUnDoIng);//触发回滚

                //更新数据
                simpleRedisConfig.getRedisInterface().set(ProcessFlowHead + processFlow.getKey(),JSONUtil.toJsonStr(processFlow));

                //失败回调
                tigErr(processFlow.getKey(),processWorkerReturn.getReturnOrFailMsg());

                simpleDistributedLock.dLockRemove(DealLockHead+processFlow.getKey());
                tigNextStep(processFlow.getKey());//下一步触发
            }

        }

    }

    /**
     * 回滚流程处理
     *
     * @param processFlow
     * @param workerInterface
     */
    private void unDoingWork(ProcessFlowMsgBean processFlow, SimpleProcessFlowWorkerInterface workerInterface) {

        ProcessWorkerReturn processWorkerReturn =new ProcessWorkerReturn(ProcessWorkerReturn.StateRetry);

        try{

            String returnInfo=null;
            if(processFlow.getCurIndex()<processFlow.getReturnInfo().size()){
                returnInfo=processFlow.getReturnInfo().get(processFlow.getCurIndex());
            }

            processWorkerReturn =workerInterface.unDoWork(
                    JSONUtil.toList(processFlow.getParamInfo().get(processFlow.getCurIndex()), String.class)
                    ,returnInfo);

            if(processWorkerReturn ==null){
                processWorkerReturn =new ProcessWorkerReturn(ProcessWorkerReturn.StateSuc);
            }

            if(ProcessWorkerReturn.StateSuc.equals(processWorkerReturn.getState())){

                if(processFlow.getCurIndex()>0){//未走到终点
                    processFlow.setCurIndex(processFlow.getCurIndex() - 1);
                }else {
                    processFlow.setState(ProcessFlowMsgBean.StateUnDoed);
                }
            }

        }finally {


            if(ProcessWorkerReturn.StateSuc.equals(processWorkerReturn.getState())){

                if(ProcessFlowMsgBean.StateUnDoed.equals(processFlow.getState())){//已走到终点

                    simpleRedisConfig.getRedisInterface().delete(ProcessFlowHead+processFlow.getKey());
                    simpleRedisConfig.getRedisInterface().set(ProcessFlowRecordHead+processFlow.getKey()
                            ,JSONUtil.toJsonStr(processFlow),Duration.ofMillis(RecordDeadTime));

                }else {
                    //更新数据
                    simpleRedisConfig.getRedisInterface().set(ProcessFlowHead + processFlow.getKey()
                            ,JSONUtil.toJsonStr(processFlow));

                    simpleDistributedLock.dLockRemove(DealLockHead+processFlow.getKey());
                    tigNextStep(processFlow.getKey());//下一步触发
                }
            }else if(ProcessWorkerReturn.StateRetry.equals(processWorkerReturn.getState())){

                delayList.add(processFlow.getKey());
                simpleDistributedLock.dLockRemove(DealLockHead+processFlow.getKey());
                tigNextStep(processFlow.getKey());//下一步触发

            }else {//有节点处理失败的

                //回滚中并且处理失败//触发死信
                processFlow.setState(ProcessFlowMsgBean.StateDead);

                //新增错误信息
                JSONObject jsonObject=new JSONObject();
                jsonObject.set("msg", processWorkerReturn.getReturnOrFailMsg());
                jsonObject.set("errIndex",processFlow.getCurIndex());
                jsonObject.set("errState", processWorkerReturn.getState());
                jsonObject.set("oriData",processFlow);

                simpleRedisConfig.getRedisInterface().set(ProcessFlowDeadHead + processFlow.getKey()
                        ,jsonObject.toString(),
                        Duration.ofMillis(1000L*3600*24*15));

                //删除原始信息

                simpleRedisConfig.getRedisInterface().delete(ProcessFlowHead+processFlow.getKey());
                simpleRedisConfig.getRedisInterface().set(ProcessFlowRecordHead+processFlow.getKey()
                        ,JSONUtil.toJsonStr(processFlow),Duration.ofMillis(RecordDeadTime));

                simpleDistributedLock.dLockRemove(DealLockHead+processFlow.getKey());

            }
        }


    }

    public void dealListenerTimeOut(String key){
        String realKey=key.split(":")[1];

        SimpleProcessFlowListener simpleProcessFlowListener = listenerHashMap.get(realKey);

        if(simpleProcessFlowListener==null){
            return;
        }

        System.out.println("SimpleProcess 处理 监听器兜底"+key);

        String value = simpleRedisConfig.getRedisInterface().getValue(ProcessFlowRecordHead + key);

        ProcessFlowMsgBean processFlow = JSONUtil.toBean(value, ProcessFlowMsgBean.class);

        if(ProcessFlowMsgBean.StateSucc.equals(processFlow.getState())){
            simpleProcessFlowListener.onSucc(realKey, processFlow.getReturnInfo());

        }else {

            simpleProcessFlowListener.onErr(realKey,"Listener time out unKnow");

        }


    }

    private void tigNextStep(String key){

        simpleRedisConfig.getRedisInterface().convertAndSend(
                simpleRedisConfig.getKeyHead()+ simpleRedisConfig.getProduct()
                ,ProcessFlowTigHead + key);

    }
    private void tigDoing(String key,Integer curIndex,String workTypeName){

        simpleRedisConfig.getRedisInterface().convertAndSend(
                simpleRedisConfig.getKeyHead()+ simpleRedisConfig.getProduct(),
                ProcessFlowDoingTigHead + key+":"
                        +curIndex+":"
                        +workTypeName
        );

    }
    private void tigSucc(String key){

        simpleRedisConfig.getRedisInterface().convertAndSend(
                simpleRedisConfig.getKeyHead()+ simpleRedisConfig.getProduct()
                ,ProcessFlowSuccTigHead + key
        );
    }

    private void tigErr(String key,String errMsg){

        simpleRedisConfig.getRedisInterface().convertAndSend(
                simpleRedisConfig.getKeyHead()+ simpleRedisConfig.getProduct()
                ,ProcessFlowErrTigHead + key+":"+ errMsg
        );

    }


}
